package day06;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;
import po.OrderEvent;

/**
 * @Description: 使用水位线模拟超时对账失败
 * @Author ZYX
 * @Date 2022/2/15 21:01
 * @Version 1.0
 */
public class Demo03 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<OrderEvent> orderEventStream = env.addSource(
                new SourceFunction<OrderEvent>() {
                    @Override
                    public void run(SourceContext<OrderEvent> sourceContext) throws Exception {
                        sourceContext.collectWithTimestamp(OrderEvent.of("order-1","order",1000L),1000L);
                        sourceContext.emitWatermark(new Watermark(999L));
                        sourceContext.collectWithTimestamp(OrderEvent.of("order-2","order",2000L),2000L);
                        sourceContext.emitWatermark(new Watermark(7000L));
                    }

                    @Override
                    public void cancel() {

                    }
                }
        );

        SingleOutputStreamOperator<OrderEvent> wechatStream = env.addSource(
                new SourceFunction<OrderEvent>() {
                    @Override
                    public void run(SourceContext<OrderEvent> sourceContext) throws Exception {
                        sourceContext.collectWithTimestamp(OrderEvent.of("order-1","wechat",4000L),4000L);
                        sourceContext.emitWatermark(new Watermark(3999L));
                        sourceContext.emitWatermark(new Watermark(7000L));
                        sourceContext.collectWithTimestamp(OrderEvent.of("order-2","wechat",8000L),8000L);

                    }

                    @Override
                    public void cancel() {

                    }
                }
        );

        orderEventStream.keyBy(obj -> obj.orderId)
                .connect(wechatStream.keyBy(obj -> obj.orderId))
                .process(new MatchFunction())
                .print();


        env.execute();
    }

    public static class MatchFunction extends CoProcessFunction<OrderEvent,OrderEvent,String>{
        private ValueState<OrderEvent> orderState;
        private ValueState<OrderEvent> wechatState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            orderState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("order", Types.POJO(OrderEvent.class)));
            wechatState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("wechat", Types.POJO(OrderEvent.class)));

        }

        @Override
        public void processElement1(OrderEvent orderEvent, Context context, Collector<String> out) throws Exception {
            if (wechatState.value() == null){
                // 下订单order事件先到达，因为如果weixin事件先到达，那么weixinState就不为空了
                orderState.update(orderEvent);
                context.timerService().registerEventTimeTimer(orderEvent.timestamp + 5000L);
            }else {
                out.collect("订单id是："+orderEvent.orderId+"对账成功，wechat事件先到达");
                wechatState.clear();
            }
        }

        @Override
        public void processElement2(OrderEvent wechatEvent, Context context, Collector<String> out) throws Exception {
            if (orderState.value() == null){
                wechatState.update(wechatEvent);
                context.timerService().registerEventTimeTimer(wechatEvent.timestamp+5000L);
            }else {
                out.collect("订单id是"+wechatEvent.orderId+"对账成功，order事件先到达");
                orderState.clear();
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            if (orderState.value() != null) {
                out.collect("订单ID" + orderState.value().orderId + "对账失败，weixin事件5s内未到达");
                orderState.clear();
            }
            if (wechatState.value() != null) {
                out.collect("订单ID" + wechatState.value().orderId + "对账失败，order事件5s内未到达");
                wechatState.clear();
            }
        }
    }

}
